package ex;

import org.apache.commons.io.IOUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;

public class BoltMatching extends BaseRichBolt {
    private OutputCollector collector;
    private final Hashtable<String, String> list = new Hashtable<>();
    private Set<String> keys = null;

    private void addKeywords(String filename, String state) {
        InputStream inputStream = ClassLoader.getSystemClassLoader().getResourceAsStream(filename);
        try {
            assert inputStream != null;
            List<String> strings = IOUtils.readLines(inputStream, "UTF-8");
            // strings.for /*一种快速写for each的方式*/
            for (String key : strings) {
                list.put(key,state);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        collector = outputCollector;
        addKeywords("important.txt", PublicSentimentTopology.STATE_IMPORTANT);
        addKeywords("focus.txt", PublicSentimentTopology.STATE_FOCUS);
        addKeywords("hot.txt", PublicSentimentTopology.STATE_HOT);
        keys = list.keySet();
    }

    @Override
    public void execute(Tuple tuple) {
        String info = tuple.getStringByField("info");

        for (String key: keys) {
            if (info.contains(key)) {
                collector.emit(list.get(key), new Values(info));
                return;
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declareStream(PublicSentimentTopology.STATE_IMPORTANT, new Fields("info"));
        outputFieldsDeclarer.declareStream(PublicSentimentTopology.STATE_FOCUS, new Fields("info"));
        outputFieldsDeclarer.declareStream(PublicSentimentTopology.STATE_HOT, new Fields("info"));
    }
}
